package com.action
import java.sql.Timestamp
import java.util.{Timer, TimerTask}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, Row, SparkSession}
import java.sql.{Connection, DriverManager, PreparedStatement}


import java.util.Properties

object TopAndSave {
  // 创建 SparkSession
  val spark = SparkSession.builder()
    .appName("als-demo")
    .master("local[*]")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .getOrCreate()

  // 从 MySQL 数据库读取数据
  val props = new Properties()
  props.setProperty("user", "root")
  props.setProperty("password", "123456")
  val jdbcUrl = "jdbc:mysql://localhost:3306/hf-test"

  def getCommDataframe(): RDD[(Long, Long, Double, Timestamp)] = {
    val query = "(SELECT user_id as userId, trend_id as trendId,create_time as createTime FROM trend_comment) AS trend_comment_tmp"
    val userId_trendId: DataFrame = spark.read.jdbc(jdbcUrl, query, props)
    val like_rating = 1.0
    val rating: RDD[(Long, Long, Double,Timestamp)] = userId_trendId.rdd.map {
      case Row(userId: Long, trendId: Long ,createTime:Timestamp) => (userId, trendId, like_rating, createTime)
    }
    rating
  }


  def getLikeDataframe(): RDD[(Long, Long, Double, Timestamp)] = {

    val query = "(SELECT user_id as userId, trend_id as trendId,create_time as createTime FROM action_user_trend_like) AS trend_like_tmp"
    val userId_trendId: DataFrame = spark.read.jdbc(jdbcUrl, query, props)
    val like_rating = 2.0
    val rating: RDD[(Long, Long, Double,Timestamp)] = userId_trendId.rdd.map {
      case Row(userId: Long, trendId: Long,createTime:Timestamp) => (userId, trendId, like_rating, createTime)
    }
    rating
  }


  def getSaveDataframe(): RDD[(Long, Long, Double, Timestamp)] = {
    val query = "(SELECT user_id as userId, trend_id as trendId,create_time as createTime FROM action_user_trend_save) AS trend_save_tmp"
    val userId_trendId: DataFrame = spark.read.jdbc(jdbcUrl, query, props)
    val like_rating = 3.0
    val rating: RDD[(Long, Long, Double, Timestamp)] = userId_trendId.rdd.map {
      case Row(userId: Long, trendId: Long,createTime:Timestamp) => (userId, trendId, like_rating, createTime)
    }
    rating
  }

  // 定义一个方法来合并所有行为的 RDD
  def mergeBehaviorDataframes(): RDD[(Long, (Long, Long, Double, Timestamp))] = {
    // 获取所有行为的 RDD
    val commDataframe = getCommDataframe()
    val likeDataframe = getLikeDataframe()
    val saveDataframe = getSaveDataframe()


    // 使用 union 方法将这些 RDD 合并为一个 RDD
    val allRDD = commDataframe.union(likeDataframe).union(saveDataframe)

    // 按 trendId 进行分组
    val groupedRDD: RDD[(Long, (Long, Long, Double, Timestamp))] = allRDD.map {
      case (userId, trendId, rating, createTime) => (trendId, (userId, trendId, rating, createTime))
    }

    groupedRDD
  }

  def countLikeAndSave():RDD[(Long,(Int, Int, Int))] ={
    val groupedRDD = mergeBehaviorDataframes()

    // 对每组的数据进行分组和统计
    val countedRDD: RDD[(Long,(Int, Int, Int))] = groupedRDD.groupByKey().mapValues { iter =>
      var topCount = 0
      var commCount = 0
      var saveCount = 0

      iter.foreach {
        case (_, _, rating, _) =>
          if (rating == 2.0) topCount += 1
          else if (rating == 1.0) commCount += 1
          else if (rating == 3.0) saveCount += 1
      }

      (topCount, commCount, saveCount)
    }
    countedRDD
  }

  // 定义方法来更新 trend 表中的数据
  def updateTrendTable(resultRDD: RDD[(Long, (Int, Int, Int))]): Unit = {
    // MySQL 连接配置
    val jdbcUrl = "jdbc:mysql://localhost:3306/hf-test"
    val user = "root"
    val password = "123456"

    // 将结果 RDD 转换为需要的 Map 数据结构
    val trendUpdates: Map[Long, (Int, Int, Int)] = resultRDD.collect().toMap.view.mapValues {
      case (newTopCount, newComCount, newSaveCount) => (newTopCount, newComCount, newSaveCount)
    }.toMap

    // 调用 updateTrendData 方法执行更新
    updateTrendData(jdbcUrl, user, password, trendUpdates)
  }

  def updateTrendData(url: String, user: String, password: String, trendUpdates: Map[Long, (Int, Int, Int)]): Unit = {
    var connection: Connection = null
    var preparedStatement: PreparedStatement = null

    try {
      // 连接数据库
      connection = DriverManager.getConnection(url, user, password)

      // 构建更新 SQL
      val sql = "UPDATE trend SET top_count = ?, com_count = ?, save_count = ? WHERE trend_id = ?"
      preparedStatement = connection.prepareStatement(sql)

      // 设置自动提交为 false，以便手动控制事务
      connection.setAutoCommit(false)

      // 遍历需要更新的博客数据并添加到批处理中
      trendUpdates.foreach { case (trendId, (newTopCount, newComCount, newSaveCount)) =>
        preparedStatement.setInt(1, newTopCount)
        preparedStatement.setInt(2, newComCount)
        preparedStatement.setInt(3, newSaveCount)
        preparedStatement.setLong(4, trendId)
        preparedStatement.addBatch() // 添加到批处理中
      }

      // 执行批量更新
      preparedStatement.executeBatch()

      // 手动提交事务
      connection.commit()

      println("Batch update successful!")

    } catch {
      case e: Exception =>
        // 发生异常时回滚事务
        if (connection != null) connection.rollback()
        println("Error occurred while updating trend data: " + e.getMessage)
    } finally {
      // 恢复自动提交
      if (connection != null) connection.setAutoCommit(true)

      // 关闭连接和 preparedStatement
      if (preparedStatement != null) preparedStatement.close()
      if (connection != null) connection.close()
    }
  }


  def begin():Unit={
    // 示例用法：统计每个博客的点赞数量、评论数量和收藏数量
    val resultRDD: RDD[(Long, (Int, Int, Int))] = countLikeAndSave()

    // 更新 trend 表中的数据
    updateTrendTable(resultRDD)

  }

  def main(args: Array[String]): Unit = {

    // 创建定时器对象
    val timer = new Timer()

    // 定义定时任务
    val task = new TimerTask {
      def run(): Unit = {
        // 在这里执行你想要定时执行的任务
        println("Executing task...")

        // 调用你的需要执行的函数
        begin()

        // 这里是示例代码，可以替换为你的逻辑

        println("Task completed.")
      }
    }

    // 每隔10分钟执行一次任务，初始延迟为0，间隔为10分钟（10 * 60 * 1000 毫秒）
    timer.schedule(task, 0,  24 * 60 * 60 * 1000)



  }

}
